package com.atguigu.edu.realtime.function;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.LogTagType;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName: LogSplitProcessFunction
 * @Description: TODO 对流中数据按照日志规则进行分流处理
 * @Author: zhaoxunfeng
 * @Date: 2022-08-31 14:32
 * @Version: 1.0.0
 */
public class LogSplitProcessFunction extends ProcessFunction<JSONObject, JSONObject> {
    @Override
    public void processElement(JSONObject json,
                               Context ctx,
                               Collector<JSONObject> out) throws Exception {

        // 处理错误日志
        if (json.getJSONObject("err") != null) {
            ctx.output(LogTagType.ERR_LOG_TAG, json);
            // 去除数据中的err的内容
            json.remove("err");
        }

        // 处理启动日志
        if (json.getJSONObject("start") != null) {
            ctx.output(LogTagType.START_LOG_TAG, json);
        } else if (json.getJSONObject("appVideo") != null) {
            // 处理视频播放日志
            ctx.output(LogTagType.PLAY_LOG_TAG, json);
        } else {

            JSONObject common = json.getJSONObject("common");
            JSONObject page = json.getJSONObject("page");
            Long ts = json.getLong("ts");

            // 处理曝光日志
            JSONArray displays = json.getJSONArray("displays");
            if (displays != null) {
                for (Object display : displays) {
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("common", common);
                    jsonObject.put("display", display);
                    jsonObject.put("page", page);
                    jsonObject.put("ts", ts);
                    ctx.output(LogTagType.DISPLAY_LOG_TAG, jsonObject);
                }
                // 去除日志中的displays标签
                json.remove("displays");
            }


            // 处理动作行为日志
            JSONArray actions = json.getJSONArray("actions");
            if (actions != null) {
                for (Object action : actions) {
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("common", common);
                    jsonObject.put("action", action);
                    jsonObject.put("page", page);
                    jsonObject.put("ts", ts);
                    ctx.output(LogTagType.ACTION_LOG_TAG, jsonObject);
                }
                // 去除日志中的actions标签
                json.remove("actions");
            }

            // 日志中剩余的数据就是页面浏览日志
            ctx.output(LogTagType.PAGE_LOG_TAG, json);
        }
    }
}
